package com.hfzy.ihk.web.ivr.context;
import com.hfzy.ihk.web.ivr.constant.PublicConfig;
import com.hfzy.ihk.web.ivr.support.callin.event.CallIntoEvent;
import com.hfzy.ihk.web.ivr.support.callin.event.FreeswitchCallOutEvent;
import com.hfzy.ihk.web.ivr.support.callin.event.FreeswitchCallStateEvent;
import com.hfzy.ihk.web.ivr.support.callin.factory.CallIntoEventFactory;
import com.hfzy.ihk.web.ivr.support.callin.factory.FreeswitchCallOutEventFactory;
import com.hfzy.ihk.web.ivr.support.callin.factory.FreeswitchCallStateFactory;
import com.hfzy.ihk.web.ivr.support.callin.handler.*;
import com.hfzy.ihk.web.ivr.support.disruptor.exception.DisruptorExceptionHandler;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Administrator on 16-5-28.
 */

@Component("disruptorInitBean")
public class DisruptorInitBean implements InitializingBean, DisposableBean {

    private Logger logger = LoggerFactory.getLogger(DisruptorInitBean.class);


    //Freeswitch呼入队列
    private static RingBuffer<CallIntoEvent> ringBufferFreeswitchCallInto;
    private static RingBuffer<CallIntoEvent> ringBufferDongjinCallInto;
    //freeswitch状态队列
    public static RingBuffer<FreeswitchCallStateEvent>[] ringBufferFreeswitchCallState=new RingBuffer[PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE];
    //freeswitch单呼说话状态队列
    public static RingBuffer<FreeswitchCallStateEvent>[] ringBufferFreeswitchAsrCallSpeakState=new RingBuffer[PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE];

    //Freeswitch呼出队列
    private static RingBuffer<FreeswitchCallOutEvent> ringBufferFreeswitchCallOut;


    private Disruptor<CallIntoEvent> disruptorFreeswitchCallInto;
    private Disruptor<CallIntoEvent> disruptorDongjinCallInto;
    private Disruptor<FreeswitchCallOutEvent> disruptorFreeswitchCallOut;
    private Disruptor<FreeswitchCallStateEvent> disruptorFreeswitchCallState;
    private Disruptor<FreeswitchCallStateEvent> disruptorFreeswitchAsrCallSpeakState;


    //线程
    private ExecutorService executorFreeswitchCallInto;
    private ExecutorService executorDongjinCallInto;
    private ExecutorService executorFreeswitchCallOut;
    private ExecutorService executorFreeswitchCallState;
    private ExecutorService executorFreeswitchAsrCallSpeakState;

    public static RingBuffer<CallIntoEvent> getRingBufferFreeswitchCallInto() {
        return ringBufferFreeswitchCallInto;
    }

    public static RingBuffer<CallIntoEvent> getRingBufferDongjinCallInto() {
        return ringBufferDongjinCallInto;
    }

    public static RingBuffer<FreeswitchCallStateEvent> getRingBufferFreeswitchCallState(int num){
        return  ringBufferFreeswitchCallState[num];
    }

    public static RingBuffer<FreeswitchCallOutEvent> getRingBufferFreeswitchCallOut() {
        return ringBufferFreeswitchCallOut;
    }

    public static RingBuffer<FreeswitchCallStateEvent> getRingBufferFreeswitchAsrCallSpeakState(int num) {
        return ringBufferFreeswitchAsrCallSpeakState[num];
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        logger.info("ivr开始初始化云呼队列");
        initDisruptorFreeswitchCallInto();
        initDisruptorDongjinCallState();
        initDisruptorFreeswitchCallOut();
        initDisruptorFreeswitchCallState();
        initDisruptorFreeswitchAsrCallSpeakState();
        logger.info("ivr初始化云呼队列完成");
    }

    /**
     * 初始化freeswitch外呼队列
     */
    private void initDisruptorFreeswitchCallInto(){
        int handlerSize = PublicConfig.FS_CALL_INTO_HANDLER_SIZE;
        int bufferSize = PublicConfig.FS_CALL_INTO_BUFFER_SIZE;

        executorFreeswitchCallInto = Executors.newFixedThreadPool(handlerSize);

        CallIntoEventFactory callEventFactory = new CallIntoEventFactory();

        disruptorFreeswitchCallInto = new Disruptor(
                callEventFactory,
                bufferSize,
                executorFreeswitchCallInto,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        FreeswitchCallIntoHandler[] handlers = new FreeswitchCallIntoHandler[handlerSize];
        for (int i = 0; i < handlerSize; i++) {
            handlers[i] = BeanUtils.getFreeswitchCallIntoHandler();
        }

        disruptorFreeswitchCallInto.handleEventsWithWorkerPool(handlers);

        disruptorFreeswitchCallInto.setDefaultExceptionHandler(new DisruptorExceptionHandler());

        ringBufferFreeswitchCallInto = disruptorFreeswitchCallInto.start();

        logger.info("ivr init disruptorFreeswitchCallInto over: handler size:{},buffer size:{}", handlerSize, bufferSize);
    }

    private void initDisruptorDongjinCallState() {

        int handlerSize = PublicConfig.DONGJIN_CALL_STATE_HANDLER_SIZE;
        int bufferSize  = PublicConfig.DONGJIN_CALL_STATE_BUFFER_SIZE;

        executorDongjinCallInto = Executors.newFixedThreadPool(handlerSize);

        CallIntoEventFactory callEventFactory = new CallIntoEventFactory();

        disruptorDongjinCallInto = new Disruptor(
                callEventFactory,
                bufferSize,
                executorDongjinCallInto,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        DongjinCallIntoHandler handlers[] = new DongjinCallIntoHandler[handlerSize];
        for(int i=0;i<handlerSize;i++){
            handlers[i] = BeanUtils.getDongjinCallIntoHandler();
        }

        disruptorDongjinCallInto.handleEventsWithWorkerPool(handlers);

        ringBufferDongjinCallInto = disruptorDongjinCallInto.start();

        logger.info("ivr init disruptorDongjinCallInto over: handler size:{},buffer size:{}",handlerSize,bufferSize);
    }

    /**
     * 初始化freeswitch状态队列
     */
    private void initDisruptorFreeswitchCallState(){

        int handlerSize=PublicConfig.ALIYUN_CALL_STATE_HANDLER_SIZE;

        int bufferSize=PublicConfig.ALIYUN_CALL_STATE_BUFFER_SIZE;

        int bufferNumSize = PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE;

        for(int a=0;a<bufferNumSize;a++){

            executorFreeswitchCallState=Executors.newFixedThreadPool(handlerSize);

            FreeswitchCallStateFactory factory=new FreeswitchCallStateFactory();

            disruptorFreeswitchCallState=new Disruptor(factory,
                    bufferSize,
                    executorFreeswitchCallState,
                    ProducerType.MULTI,new BlockingWaitStrategy());

            FreeswitchAsrCallStateHandler[] handlers=new FreeswitchAsrCallStateHandler[handlerSize];

            for (int b=0;b<handlers.length;b++){

                handlers[b]=BeanUtils.getFreeswitchAsrCallStateHandler();
            }

            disruptorFreeswitchCallState.handleEventsWithWorkerPool(handlers);

            disruptorFreeswitchCallState.setDefaultExceptionHandler(new DisruptorExceptionHandler());

            ringBufferFreeswitchCallState[a]=disruptorFreeswitchCallState.start();
        }

        logger.info("ivr init disruptorFreeswitchCallState over: handler size:{},buffer size:{}", handlerSize, bufferSize);


    }

    /**
     * 初始化freeswitch外呼队列
     */
    private void initDisruptorFreeswitchCallOut(){
        int handlerSize = PublicConfig.ALIYUN_CALL_OUT_HANDLER_SIZE;
        int bufferSize = PublicConfig.ALIYUN_CALL_OUT_BUFFER_SIZE;

        executorFreeswitchCallOut = Executors.newFixedThreadPool(handlerSize);

        FreeswitchCallOutEventFactory callEventFactory = new FreeswitchCallOutEventFactory();

        disruptorFreeswitchCallOut = new Disruptor(
                callEventFactory,
                bufferSize,
                executorFreeswitchCallOut,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        FreeswitchAsrCallOutHandler[] handlers = new FreeswitchAsrCallOutHandler[handlerSize];
        for (int i = 0; i < handlerSize; i++) {
            handlers[i] = BeanUtils.getFreeswitchAsrCallOutHandler();
        }

        disruptorFreeswitchCallOut.handleEventsWithWorkerPool(handlers);

        disruptorFreeswitchCallOut.setDefaultExceptionHandler(new DisruptorExceptionHandler());

        ringBufferFreeswitchCallOut = disruptorFreeswitchCallOut.start();

        logger.info("ivr init disruptorFreeswitchCallOut over: handler size:{},buffer size:{}", handlerSize, bufferSize);
    }

    /**
     * 初始化freeswitch状态队列
     */
    private void initDisruptorFreeswitchAsrCallSpeakState(){

        int handlerSize=PublicConfig.ALIYUN_CALL_STATE_HANDLER_SIZE;

        int bufferSize=PublicConfig.ALIYUN_CALL_STATE_BUFFER_SIZE;

        int bufferNumSize = PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE;

        for(int a=0;a<bufferNumSize;a++){

            executorFreeswitchAsrCallSpeakState=Executors.newFixedThreadPool(handlerSize);

            FreeswitchCallStateFactory factory=new FreeswitchCallStateFactory();

            disruptorFreeswitchAsrCallSpeakState=new Disruptor(factory,
                    bufferSize,
                    executorFreeswitchAsrCallSpeakState,
                    ProducerType.MULTI,new BlockingWaitStrategy());

            FreeswitchAsrCallSpeakStateHandler[] handlers=new FreeswitchAsrCallSpeakStateHandler[handlerSize];

            for (int b=0;b<handlers.length;b++){

                handlers[b]=BeanUtils.getFreeswitchAsrCallSpeakStateHandler();
            }

            disruptorFreeswitchAsrCallSpeakState.handleEventsWithWorkerPool(handlers);

            disruptorFreeswitchAsrCallSpeakState.setDefaultExceptionHandler(new DisruptorExceptionHandler());

            ringBufferFreeswitchAsrCallSpeakState[a]=disruptorFreeswitchAsrCallSpeakState.start();
        }

        logger.info("ivr init disruptorFreeswitchAsrCallSpeakState over: handler size:{},buffer size:{}", handlerSize, bufferSize);


    }

    @Override
    public void destroy() throws Exception {
        logger.info("ivr disruptor destroy..............");

        //关闭freeswitch相关
        disruptorFreeswitchCallInto.shutdown();
        disruptorDongjinCallInto.shutdown();
        disruptorFreeswitchCallOut.shutdown();
        disruptorFreeswitchCallState.shutdown();
        disruptorFreeswitchAsrCallSpeakState.shutdown();

        executorDongjinCallInto.shutdown();
        executorFreeswitchCallInto.shutdown();
        executorFreeswitchCallOut.shutdown();
        executorFreeswitchCallState.shutdown();
        executorFreeswitchAsrCallSpeakState.shutdown();
    }

}
